package com.dataaccess.dataaccess.config.rocketmq;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.dataaccess.dataaccess.support.udp.DataReceiveConfigration;
import com.dataaccess.dataaccess.support.utils.MyLogger;
import com.dataaccess.dataaccess.web.service.*;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @Author 18011618
 * @Date 10:31 2018/7/18
 * @Function 模拟用户消息消费
 */
@Component
public class UserConsumer {
    /**
     * 消费者的组名
     */
    @Value("${suning.rocketmq.conumerGroup}")
    private String consumerGroup;

    /**
     * NameServer 地址
     */
    @Value("${suning.rocketmq.namesrvaddr}")
    private String namesrvAddr;

    @Resource
    public MyLogger logger;

    @Resource
    public DeviceAboatCombustiblegasService deviceAboatCombustiblegasService;
    @Resource
    public DeviceAboatElectricService deviceAboatElectricService;
    @Resource
    private DeviceAboatFireExtinguisherService deviceAboatFireExtinguisherService;
    @Resource
    public DeviceAboatFirewaterService deviceAboatFirewaterService;
    @Resource
    public DeviceAboatSmokeService deviceAboatService;
    @Resource
    private DeviceAboatGasService deviceAboatGasService;
    @Resource
    private VisualizationVideoAlarmService videoAlarmService;
    @Resource
    DataReceiveConfigration dataReceiveConfigration;
    @Resource
    PostOrGetService postOrGetService;
    @Resource
    SatwellWirelessGatewayService satwellWirelessGatewayService;
    @Resource
    private WaterPumpService waterPumpService;
    @Resource
    DeviceOpenElectricService deviceOpenElectricService;
    @Resource
    AutoNetworkAlarmDeviceDataChangedService autoNetworkAlarmDeviceDataChangedService;
    @Resource
    private SanJiangYunService sanJiangYunService;
    @Resource
    private DeviceSwitchService deviceSwitchService;
    @PostConstruct
    public void consumer() {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(namesrvAddr);
        consumer.setConsumeMessageBatchMaxSize(100);
        consumer.setPullBatchSize(100);
        try {
            consumer.subscribe("user-topic", "user-tag");
            //注册消息监听器处理消息
//            consumer.registerMessageListener(new MessageListenerConcurrently() {
//                @Override
//                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context){
//                    //对消息进行处理
//                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
//                }
//            });
            consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
                try {
                    for (MessageExt messageExt : list) {
                        String mes = new String(messageExt.getBody());
                        logger.info("consumer" + "------" + mes);
                        JSONObject jsonObject = JSONUtil.parseObj(mes);
                        String deviceFlag = jsonObject.getStr("deviceFlag");
                        if (deviceFlag.equals("udpCombustiblegas")) {
                            dataReceiveConfigration.handleMes(mes, "udp");
                        } else if (deviceFlag.equals("combustiblegas")) {
                            deviceAboatCombustiblegasService.deviceDataChanged(jsonObject);
                        } else if (deviceFlag.equals("electric")) {
                            try {
                                deviceAboatElectricService.deviceDataChanged(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("openelectric")){
                            try{
                                deviceOpenElectricService.openElectric(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("scdqElectricTelecom")){
                            try{
                                deviceAboatElectricService.scdqElectricTelecom(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("shengshanElectricTelecom")){
                            try{
                                deviceAboatElectricService.shengshanTelecom(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("hkElectricTelecom")) {
                            try {
                                deviceAboatElectricService.HK641(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("shengshanElectricTCP")){
                            try{
                                deviceAboatElectricService.shengshanTCP(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("electricHik")){
                            try{
                                deviceAboatElectricService.electricHik(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("fireExtinguisher")) {
                            try {
                                deviceAboatFireExtinguisherService.deviceDataChanged(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("gasFireExtinguishingIotData")) {
                            try {
                                deviceAboatFireExtinguisherService.gasFireExtinguishingIotData(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        } else if (deviceFlag.equals("firewater")) {
                            try {
                                deviceAboatFirewaterService.deviceDataChanged(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("shichengWaterDX")) {
                            try {
                                deviceAboatFirewaterService.shichengDX(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("shichengDX_new")) {
                            try {
                                deviceAboatFirewaterService.shichengDX_new(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("wirelessInstrument")) {
                            try {
                                deviceAboatFirewaterService.wirelessInstrument(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("waterPump")){
                            try{
                                waterPumpService.deviceDataChanged(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        } else if (deviceFlag.equals("SCDQLORA")) {
                            try {
                                deviceAboatFirewaterService.scdqLORA(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        } else if (deviceFlag.equals("smoke")) {
                            deviceAboatService.deviceDataChanged(jsonObject);
                        }else if(deviceFlag.equals("siterwellTMSmoke")) {
                            try {
                                deviceAboatService.siterWell(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        } else if(deviceFlag.equals("haiKang")) {
                            try {
                                deviceAboatService.haiKang(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("chitongda")) {
                            try {
                                deviceAboatService.chitongda(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("sanJiangSmokeTelecom")) {
                            try {
                                deviceAboatService.sanJiangSmokeTelecom(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("thinKingZhiAnSmoke")) {
                            try {
                                deviceAboatService.thinKingZhiAnSmoke(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("video")) {
                            try {
                                videoAlarmService.fluoriteCloud(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("waveBox")) {
                            try {
                                videoAlarmService.waveBox(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("yunCongGasStationAnalyzer")) {
                            try {
                                videoAlarmService.yunCongGasStationAnalyzer(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("aliyunCamera")) {
                            try {
                                videoAlarmService.aliyunCamera(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("rongHeYongDao")) {
                            try {
                                videoAlarmService.rongHeYongDao(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("daHuaYunRui")){
                            try{
                                videoAlarmService.daHuaYunRui(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        } else if (deviceFlag.equals("hikvision")) {
                            try {
                                videoAlarmService.hikvision(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        } else if (deviceFlag.equals("hikvisionBounds")) {
                            try {
                                videoAlarmService.eventSubscriptionsHKRest(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("unloadingOilOrSafe")) {
                            try {
                                videoAlarmService.unloadingOilOrSafe(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        } else if (deviceFlag.equals("hkHyperbrain")) {
                            try {
                                videoAlarmService.hkHyperbrain(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("flameJDY")) {
                            try {
                                videoAlarmService.flameJDY(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if (deviceFlag.equals("updateToOff")) {
                            try {
                                postOrGetService.updateToOff(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("HX100")) {
                            try {
                                autoNetworkAlarmDeviceDataChangedService.deviceDataChangedHX100(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("autoNetDahua")) {
                            try {
                                autoNetworkAlarmDeviceDataChangedService.autoNetDahua(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("updateHEX100ToON")){
                            try{
                                postOrGetService.updateHEX100ToON(jsonObject);
                            }catch (Exception e){
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("thinKingZhiAnGas")) {
                            try {
                                deviceAboatGasService.thinKingZhiAnGas(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("pa8201Gas")) {
                            try {
                                deviceAboatGasService.pa8201Gas(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("jikaidaCarbonMonoxide")) {
                            try {
                                deviceAboatGasService.jikaidaCarbonMonoxide(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("hanWei_JY_KBR2")) {
                            try {
                                deviceAboatGasService.hanWei_JY_KBR2(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("JB_QB_F52")) {
                            try {
                                autoNetworkAlarmDeviceDataChangedService.JB_QB_F52(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("sanJiangYun")) {
                            try {
                                sanJiangYunService.sanJiangYun(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("switch")) {
                            try {
                                deviceSwitchService.deviceSwitchStateUpdate(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        } else if (deviceFlag.equals("test")) {
                            System.err.println("消费消息: " + "deviceFlag:" + deviceFlag + "-----" + mes);
                        } else if (deviceFlag.equals("gas")) {
                            deviceAboatGasService.panAnCO(jsonObject);
                        } else if (deviceFlag.equals("siTerWell")) {
                            try {
                                deviceAboatGasService.siTerWell(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }else if(deviceFlag.equals("GSM03PRO")) {
                            try {
                                satwellWirelessGatewayService.GSM03PRO(jsonObject);
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                        }
                        //System.err.println("消费消息: " + new String(messageExt.getBody()));//输出消息内容
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER; //稍后再试
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; //消费成功
            });
            consumer.start();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
